#include "common.h"
#include "cuckoo.h" //cuckoo hashmap
#include <jerasure.h>
#include <cauchy.h>

//======================global var====================
typedef struct WorkerAsClient2WorkerConnect  //当前节点连接其他节点的套接字  当前节点作为客户端
{
   int targetNode;//目标服务器节点
   int sockfd;   //当前节点作为客户端套接字
}WorkerAsClient2WorkerConnect;

typedef struct WorkerAsServer2WorkerAccept
{
   int sourceNode;//客户端节点
   int sockfd;  //服务器端套接字
}WorkerAsServer2WorkerAccept;


Node* nodes=NULL;//节点信息
Rack* racks=NULL;//机架信息
int nodeID=-1;//当前节点的ID
cuckoo* pool=NULL;

pthread_rwlock_t lock4Pool;

int isDestory=0;
int client2worker_sockfd[ClientNum]={-1};//accept  接受来自client的信息
int master2worker_sockfd=-1;//accept  接受来自master的信息
int worker2master_sockfd[ClientNum]={-1};//connect 给master发送消息 每个client对应一个
WorkerAsClient2WorkerConnect* workerAsClient2WorkerConnect=NULL;   //当前节点作为客户端连接的服务器端的节点信息
WorkerAsServer2WorkerAccept* workerAsServer2WorkerAccept=NULL;     //当前节点作为服务器连接的客户端的节点信息


int gateway2worker_sockfd=-1;
int worker2gateway_sockfd=-1;

int inner_rack_count=0;
int client_count=0;
int cross_rack_count=0;
FILE *inner_rack_file=NULL;
//====================================================



//====================function========================
int init_worker();  //先初始化worker. 

int handlerRequestIOFromClient(int clientID);//处理来自client的request 
int client2worker_recv(Client2WorkerRequest *request,int sockfd); 
int handlerReadRequestFromClient(long long blockID,long requestID,int clientID);
int handlerWriteRequestFromClient(long long blockID,int isPrimary,long requestID,int clientID);  
void worker2master_send(Worker2MasterRequest* request,int clientID);

int handlerRequestIOFromWorker(int workerAsClient_fd);
int worker2worker_recv(Worker2WorkerRequest *request,int workerfd);
int handlerWriteRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request);
int handlerReadRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request);
int handlerDelRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request);

int master2worker_recv(ArchivalRequest* request);
void worker2worker_send(Worker2WorkerRequest *request);

int retrieveOriginalBlockData(ArchivalRequest* archivalRequest,char** originalBlockData);
void encode4ParityBlockData(char** originalBlockData,char** parityBlockData);
void* worker2Worker_thread(void* arg);

void* archival_thread(void* arg);
void* process4client_thread(void* arg);
void destory_worker();
int lookforworker2workerfdByNodeID(int targetNodeID);

void worker2gateway_send(Worker2WorkerRequest *request);
int isInSameRack(int nodeID1,int nodeID2);
int handlerRequestIOFromGateWay();
void* process4gateway_thread(void* arg);
int gateway2worker_recv(Worker2WorkerRequest* request);
//====================================================
int init_worker(){
	printf("******************************init_worker begin ................\n");
	srand((int)time(NULL));//设置随机函数的seed
	pthread_rwlock_init(&lock4Pool,NULL);
    if (set_server_limit() <0) {
      	printf("init_worker()====set_server_limit error\n");
		return ERR;
    }
	//1.初始化配置化变量
	nodes=read_config_Node(CONFIG_NODE);
	printf("read_config_Node end............................\n");
	racks=read_config_Rack(CONFIG_RACK);
	printf("read_config_Rack end............................\n");
	nodeID = current_node(nodes);
	if(nodeID==-1) {
    	printf("node_init()====current_node is  error\n");
    	return ERR;  
    }
	printf("current_node:%d............................\n",nodeID);
	pool = cuckoo_init(100);
	printf("init_data_pool end............................\n");

	//2.初始化网络连接

	//worker作为服务端等待master的连接
	master2worker_sockfd=server_accept(Master2WorkerPort,Max_Length_backlog);
	if (master2worker_sockfd!=-1)
	{
		printf("node_init()====node accept master  successed\n"); 
	}else{
		printf("node_init()====node accept master error\n");
        return ERR;
	}

	//尝试睡眠一段时间 等待master/client 连接所有的workers
	//sleep(1);

	//worker作为客户端尝试连接master/client
	for (int i = 0; i < ClientNum; ++i)
	{
		worker2master_sockfd[i] = connect_try(nodes[NodeNumTotal-1].ip,Worker2MasterPort+nodeID);
		if (worker2master_sockfd[i]!=-1)
		{
			printf("node_init()====node[%d] connect master successed and  worker2master_sockfd=%d \n",nodeID,worker2master_sockfd[i]);
		}else{
			printf("node_init()====node[%d] connect master is error\n", nodeID);
			return ERR;
		}
	}


	

	//worker与worker之间的通信连接
	workerAsClient2WorkerConnect = (WorkerAsClient2WorkerConnect*)malloc(sizeof(WorkerAsClient2WorkerConnect)*(NodeNumTotal-2));
	workerAsServer2WorkerAccept  = (WorkerAsServer2WorkerAccept*)malloc(sizeof(WorkerAsServer2WorkerAccept)*(NodeNumTotal-2));
    int connectIndex=0;//定义索引
    int acceptIndex=0;//定义索引

	int i;
	for (i = 0; i < NodeNumTotal-1; i++)
	{
		if (i==nodeID)
		{
			//connect
			int j;
			for (j = 0; j < NodeNumTotal-1; j++)
			{
				if (j!=nodeID)
				{
					int workerAsclientfd;
					workerAsclientfd = connect_try(nodes[j].ip,Worker2WorkerPort+nodeID);
					if (workerAsclientfd!=-1)
					{
						printf("node_init()====Node[%d]AsConnectSockfd connect nodes[%d] successed\n",nodeID,j);
						workerAsClient2WorkerConnect[connectIndex].targetNode=nodes[j].nodeID;
						workerAsClient2WorkerConnect[connectIndex].sockfd=workerAsclientfd;
						connectIndex++;
					}else{
						printf("node_init()====Node[%d]AsConnectSockfd connect nodes[%d] is error\n",nodeID,j);
                    	return ERR;
					}
				}
			}
		}else{
			//accpte
			int workerAsserverfd;
			workerAsserverfd=server_accept(Worker2WorkerPort+i,Max_Length_backlog);
			if (workerAsserverfd!=-1)
			{
				printf("node_init()====node[%d]AsAcceptSockfd  accept nodes[%d] successed \n",nodeID,i);
				workerAsServer2WorkerAccept[acceptIndex].sourceNode=i;
				workerAsServer2WorkerAccept[acceptIndex].sockfd=workerAsserverfd;
				acceptIndex++;
			}else{
				printf("node_init()====Node[%d]AsAcceptSockfd  accept nodes[%d] is error\n",nodeID,i);
				return ERR;
			}
		}
	}

	//========================================================================================================================
	char* gatewayip="192.168.0.50";

	worker2gateway_sockfd = connect_try(gatewayip,Worker2GateWayPort+nodeID);
	if (worker2gateway_sockfd!=-1)
	{
		printf("node_init()====Node[%d] success connect to gateway server worker2gateway_sockfd[%d]\n",nodeID,worker2gateway_sockfd);
	}else{
		printf("node_init()====Node[%d] connect to gateway server is error\n",nodeID);
		return ERR;
	}


	gateway2worker_sockfd = server_accept(GateWay2WorkerPort,Max_Length_backlog);
	if (gateway2worker_sockfd!=-1)
	{
		printf("node_init()====Node[%d] success accept gateway server gateway2worker_sockfd[%d]\n",nodeID,gateway2worker_sockfd);
	}else{
		printf("node_init()====Node[%d] accept gateway server is error\n",nodeID);
		return ERR;
	}
	//==================================================================================================================================

	//worker作为服务端等待client的连接
	for (int i = 0; i < ClientNum; ++i)
	{
		client2worker_sockfd[i]=server_accept(Client2WorkerPort,Max_Length_backlog);
	 	if (client2worker_sockfd[i]!=-1)
	 	{
	 		printf("node_init()====node accept client  successed\n"); 
	 	}else{
	 		printf("node_init()====node accept client error\n");
        	return ERR;
	 	}
	}

	inner_rack_file=fopen("inner_rack.txt","a+");

	printf("worker_init() end===========================\n");
  	return 0;
}

//销毁worker
void destory_worker(){
	pthread_rwlock_destroy(&lock4Pool);
	if (nodes!=NULL) free(nodes);
	if (racks!=NULL) free(racks);
	if (pool!=NULL) cuckoo_destroy(pool,1);

	for (int i = 0; i < ClientNum; ++i)
	{
		if(client2worker_sockfd[i]!=-1) close(client2worker_sockfd[i]);
	}
	if(master2worker_sockfd!=-1) close(master2worker_sockfd);


	for (int i = 0; i < ClientNum; ++i)
	{
		if(worker2master_sockfd[i]!=-1) close(worker2master_sockfd[i]);
	}

	

	if (workerAsClient2WorkerConnect!=NULL)
	{
		for (int i = 0; i < NodeNumTotal-2; i++){
			if (workerAsClient2WorkerConnect[i].sockfd!=-1) close(workerAsClient2WorkerConnect[i].sockfd);
		}
		free(workerAsClient2WorkerConnect);
	}


	if (workerAsServer2WorkerAccept!=NULL)
	{
		for (int i = 0; i < NodeNumTotal-2; i++){
			if (workerAsServer2WorkerAccept[i].sockfd!=-1) close(workerAsServer2WorkerAccept[i].sockfd);
		}
		free(workerAsServer2WorkerAccept);
	}

	if(gateway2worker_sockfd!=-1) close(gateway2worker_sockfd);
	if(worker2gateway_sockfd==-1) close(worker2gateway_sockfd);
	fprintf(inner_rack_file,"nodeID:%d,client_count:%d,inner_rack_count:%d,cross_rack_count:%d\n", nodeID,client_count,inner_rack_count,cross_rack_count);
	fclose(inner_rack_file);
	
}


int client2worker_recv(Client2WorkerRequest *request,int clientID){  //node 接收来自client的请求
	size_t needRecv =sizeof(Client2WorkerRequest);
	char *recvBuf=(char *)malloc(needRecv*sizeof(char));
	int len = recv_bytes(client2worker_sockfd[clientID],recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	free(recvBuf);
	return len;
}

int worker2worker_recv(Worker2WorkerRequest *request,int workerfd){
	size_t needRecv =sizeof(Worker2WorkerRequest);
	char *recvBuf=(char *)malloc(needRecv);
	int len = recv_bytes(workerfd,recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	free(recvBuf);
	return len;
}


//处理来自client的request
int handlerRequestIOFromClient(int clientID){
	Client2WorkerRequest* client2WorkerRequest = (Client2WorkerRequest*)malloc(sizeof(Client2WorkerRequest));
	int len = client2worker_recv(client2WorkerRequest,clientID);
	if (client2WorkerRequest==NULL||len<=0)
	{
		printf("recv the incorrect request len:%d\n",len);
		return ERR;
	}

	printf("worker starts to hander requestID:%ld,blockID:%lld,flag:%d\n",client2WorkerRequest->requestID,client2WorkerRequest->blockID,client2WorkerRequest->flag);
	if (nodeID!=client2WorkerRequest->nodeID)
	{
		printf("request should access node nodeID:%d, current_nodeID:%d \n",client2WorkerRequest->nodeID,nodeID);
		return ERR;
	}
	long requestID=client2WorkerRequest->requestID;
	long blockID=client2WorkerRequest->blockID;
	int flag = client2WorkerRequest->flag;
	int isPrimary = client2WorkerRequest->isPrimary;
	free(client2WorkerRequest);

	if (flag==1)
	{
		//执行write指令
		if (handlerWriteRequestFromClient(blockID,isPrimary,requestID,clientID)==-1){
			printf("requestID:%ld handlerWriteRequest is error\n", requestID);
			return ERR;
		}else{
			printf("requestID:%ld handlerWriteRequest successed\n", requestID);
		}
	}else if (flag==0)
	{
		//执行read指令
		if (handlerReadRequestFromClient(blockID,requestID,clientID)==-1){
			printf("requestID:%ld handlerReadRequest is error\n", requestID);
			return ERR;
		}else{
			printf("requestID:%ld handlerReadRequest successed\n", requestID);
		}
	}else if(flag==2){
		close(client2worker_sockfd[clientID]);
		client2worker_sockfd[clientID]=-1;
		return ERR;
	}else{
		printf("requestID:%ld unkown flag:%d\n",requestID,flag);
		return ERR;
	}
	long requestID4Check;
	recv_bytes(worker2master_sockfd[clientID],(char*)&requestID4Check,sizeof(long));
	if (requestID4Check!=requestID) return ERR;
	pthread_rwlock_unlock(&lock4Pool);

	printf("request[%ld]:client has received the requestID from master\n", requestID);

	//发送requestID给client
	send_bytes(client2worker_sockfd[clientID],(char*)&requestID,sizeof(long));
	client_count++;
	return 0;
}

//worker处理来自client的读请求	
int handlerReadRequestFromClient(long long blockID,long requestID,int clientID){
	//查询相应的block 获得数据
	//发送数据给client
	pthread_rwlock_rdlock(&lock4Pool);
	char* sendBuf;
	char* key = lltoa(blockID);
	int isExist = cuckoo_exists(pool,key);
	if (isExist==0){
		printf("can not find the block blockID:%lld\n",blockID);
		sendBuf = (char*)malloc(sizeof(char)*Block_Size);
		int len = sprintf(sendBuf,"can not find the block blockID:%lld",blockID);
		send_bytes(client2worker_sockfd[clientID],sendBuf,len);
		free(sendBuf);
		pthread_rwlock_unlock(&lock4Pool);
		return ERR;
	}
	Block* block = (Block*)cuckoo_get(pool,key);
	sendBuf = (char*)malloc(sizeof(char)*Block_Size);
	memcpy(sendBuf,block->dataBuf,sizeof(char)*Block_Size);
	send_bytes(client2worker_sockfd[clientID],sendBuf,Block_Size);
	printf("sendBuf has sent to client successfully\n");
	free(sendBuf);
	free(key);
	//发送跟新消息给master并接受master回复
	Worker2MasterRequest* request = (Worker2MasterRequest*)malloc(sizeof(Worker2MasterRequest));
	request->requestID=requestID;
	request->blockID=blockID;
	request->nodeID=nodeID;
	request->flag=0;

	worker2master_send(request,clientID);
	printf("worker2master_send has done for updating MetaData\n");
	free(request);
	return 0;
}

//worker处理来自client的写请求	
int handlerWriteRequestFromClient(long long blockID,int isPrimary,long requestID,int clientID){
	//创建一个新的block 设置id和内容
	//添加到pool中
	
	char* recvBuf = (char*)malloc(sizeof(char)*Block_Size);
	//memset(recvBuf,'a',sizeof(char)*Block_Size);
	int len = recv_bytes(client2worker_sockfd[clientID],recvBuf,Block_Size);
	printf("recvived the blockdata from client len:%d\n",len);

	Block* block = (Block*)malloc(sizeof(Block));
	block->blockID=blockID;
	memcpy(block->dataBuf,recvBuf,sizeof(char)*Block_Size);

	pthread_rwlock_wrlock(&lock4Pool);

	char* key = lltoa(blockID);
	if(cuckoo_insert(pool,key,(void*)block)==1){
		printf("block fails to insert blockID:%lld\n", blockID);
		free(recvBuf);
		pthread_rwlock_unlock(&lock4Pool);
		return ERR;
	}else{
		printf("block successes to insert blockID:%lld\n", blockID);
	}
	free(recvBuf);
	free(key);
	//发送request给master 同步blockMeta
	Worker2MasterRequest* request = (Worker2MasterRequest*)malloc(sizeof(Worker2MasterRequest));
	request->requestID=requestID;
	request->isPrimary=isPrimary;
	request->blockID=blockID;
	request->nodeID=nodeID;
	request->flag=1;
	worker2master_send(request,clientID);
	printf("worker2master_send has done for updating MetaData\n");
	free(request);
	return 0;
}

void worker2master_send(Worker2MasterRequest* request,int clientID){
	size_t needSend=sizeof(Worker2MasterRequest);
	char *sendBuf=(char*)malloc(needSend);
	memcpy(sendBuf,request,needSend); //序列化
	send_bytes(worker2master_sockfd[clientID],sendBuf,needSend); //worker发送请求给master
	free(sendBuf);
}

int gateway2worker_recv(Worker2WorkerRequest* request){
	size_t needRecv =sizeof(Worker2WorkerRequest);
	char *recvBuf=(char *)malloc(needRecv);
	int len = recv_bytes(gateway2worker_sockfd,recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	free(recvBuf);
	return len;
}

int handlerRequestIOFromGateWay(){
	Worker2WorkerRequest* request = (Worker2WorkerRequest*)malloc(sizeof(Worker2WorkerRequest));
	request->toNodeID=-1;
	int len = gateway2worker_recv(request);
	if (len<=0)
	{
		printf("recv the incorrect request len:%d\n",len);
		return ERR;
	}

	printf("worker starts to hander request from worker\n");
	if (nodeID!=request->toNodeID)
	{
		printf("request should access node nodeID:%d, current_nodeID:%d \n",request->toNodeID,nodeID);
		return ERR;
	}

	int flag = request->flag;
	if (flag==1)
	{
		//执行write指令
		if (handlerWriteRequestFromWorker(gateway2worker_sockfd,request)==-1){
			printf("request handlerWriteRequest for parityBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerWriteRequest for parityBlock successed\n");
		}
	}else if (flag==0)
	{
		//执行read指令
		if (handlerReadRequestFromWorker(gateway2worker_sockfd,request)==-1){
			printf("request handlerReadRequest for originalBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerReadRequest for originalBlock successed\n");
		}
	}else if (flag==2)
	{
		//执行del指令
		if (handlerDelRequestFromWorker(gateway2worker_sockfd,request)==-1){
			printf("request handlerDelRequest for originalBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerDelRequest for originalBlock successed\n");
		}
	}else{
		printf("request unkown flag:%d\n",flag);
		free(request);
		return ERR;
	}
	free(request);
	cross_rack_count++;
	return 0;

}

int handlerRequestIOFromWorker(int sourceNode){
	Worker2WorkerRequest* request = (Worker2WorkerRequest*)malloc(sizeof(Worker2WorkerRequest));
	int i,workerAsClient_fd;
	for (i = 0; i < NodeNumTotal-2; ++i)
	{
		if (workerAsServer2WorkerAccept[i].sourceNode==sourceNode)
		{
			workerAsClient_fd=workerAsServer2WorkerAccept[i].sockfd;
			break;
		}
	}
	request->toNodeID=-1;
	int len = worker2worker_recv(request,workerAsClient_fd);
	if (len<=0)
	{
		printf("recv the incorrect request len:%d\n",len);
		return ERR;
	}

	printf("worker starts to hander request from worker\n");
	if (nodeID!=request->toNodeID)
	{
		printf("request should access node nodeID:%d, current_nodeID:%d \n",request->toNodeID,nodeID);
		return ERR;
	}
	
	int flag=request->flag;
	if (flag==1)
	{
		//执行write指令
		if (handlerWriteRequestFromWorker(workerAsClient_fd,request)==-1){
			printf("request handlerWriteRequest for parityBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerWriteRequest for parityBlock successed\n");
		}
	}else if (flag==0)
	{
		//执行read指令
		if (handlerReadRequestFromWorker(workerAsClient_fd,request)==-1){
			printf("request handlerReadRequest for originalBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerReadRequest for originalBlock successed\n");
		}
	}else if (flag==2)
	{
		//执行del指令
		if (handlerDelRequestFromWorker(workerAsClient_fd,request)==-1){
			printf("request handlerDelRequest for originalBlock is error\n");
			free(request);
			return ERR;
		}else{
			printf("request handlerDelRequest for originalBlock successed\n");
		}
	}else{
		printf("request unkown flag:%d\n",flag);
		free(request);
		return ERR;
	}
	free(request);
	inner_rack_count++;
	return 0;
}
//处理来自worker的写指令 与处理来自client的写指令相似
int handlerWriteRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request){
	Block* block = (Block*)malloc(sizeof(Block));
	block->blockID=request->blockID;
	char* recvBuf = (char*)malloc(sizeof(char)*Block_Size);
	
	recv_bytes(workerAsClient_fd,recvBuf,Block_Size);
	memcpy(block->dataBuf,recvBuf,sizeof(char)*Block_Size);

	pthread_rwlock_wrlock(&lock4Pool);
	char* key = lltoa(request->blockID);
	if(cuckoo_insert(pool,key,(void*)block)==1){
		printf("block fails to insert blockID:%lld\n", request->blockID);
		free(recvBuf);
		pthread_rwlock_unlock(&lock4Pool);
		return ERR;
	}else{
		printf("block successes to insert blockID:%lld\n", request->blockID);
	}
	free(recvBuf);
	free(key);
	pthread_rwlock_unlock(&lock4Pool);

	long requestID = request->requestID;
	send_bytes(workerAsClient_fd,(char*)&requestID,sizeof(long));
	return 0;
}

//处理来自worker的读指令 与处理来自client的读指令相似
int handlerReadRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request){
	//查询相应的block 获得数据
	//发送数据给worker
	char* sendBuf;
	pthread_rwlock_rdlock(&lock4Pool);
	char* key = lltoa(request->blockID);
	int isExist = cuckoo_exists(pool,key);
	if (isExist==0){
		printf("can not find the block blockID:%lld\n",request->blockID);
		sendBuf = (char*)malloc(sizeof(char)*Block_Size);
		int len = sprintf(sendBuf,"can not find the block blockID:%lld",request->blockID);
		send_bytes(workerAsClient_fd,sendBuf,len);
		free(sendBuf);
		pthread_rwlock_unlock(&lock4Pool);
		return ERR;
	}
	Block* block = (Block*)cuckoo_get(pool,key);
	sendBuf = (char*)malloc(sizeof(char)*Block_Size);
	memcpy(sendBuf,block->dataBuf,sizeof(char)*Block_Size);
	send_bytes(workerAsClient_fd,sendBuf,Block_Size);
	printf("sendBuf has sent to client successfully\n");
	free(sendBuf);
	free(key);
	pthread_rwlock_unlock(&lock4Pool);

	long requestID = request->requestID;
	send_bytes(workerAsClient_fd,(char*)&requestID,sizeof(long));
	return 0;
}

int handlerDelRequestFromWorker(int workerAsClient_fd,Worker2WorkerRequest* request){
	pthread_rwlock_wrlock(&lock4Pool);
	char* key = lltoa(request->blockID);
	int isExist = cuckoo_exists(pool,key);
	if (isExist==1)
	{
		Block* temp = (Block*)cuckoo_remove(pool,key);
		if (temp!=NULL) free(temp);
	}
	printf("BlockID[%lld] has been deleted successfully\n",request->blockID);
	pthread_rwlock_unlock(&lock4Pool);

	long requestID = request->requestID;
	send_bytes(workerAsClient_fd,(char*)&requestID,sizeof(long));
	free(key);
	return 0;
}

int master2worker_recv(ArchivalRequest* request){
	size_t needRecv =sizeof(ArchivalRequest);
	char *recvBuf=(char *)malloc(needRecv);
	int len = recv_bytes(master2worker_sockfd,recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	free(recvBuf);
	return len;
}

//监听master2worker_sockfd，查收归档request，并进行归档
/*
while(1)
1.监听master2worker_sockfd，查收归档request
2.收集归档编码需要的数据块，需要使用相应的WorkerAsClient2WorkerConnect中的sock_fd
3.编码，生成校验数据
4.发送parityBLock，同样需要使用相应的WorkerAsClient2WorkerConnect中的sock_fd
5.删除originalBlock，同样需要使用相应的WorkerAsClient2WorkerConnect中的sock_fd
6.通过master2worker_sockfd发送response
7.free空间资源
*/
void* archival_thread(void* arg){
	int i;
	while(isDestory==0){
		ArchivalRequest* request = (ArchivalRequest*)malloc(sizeof(ArchivalRequest));
		request->RG_ID=-1;
		int len = master2worker_recv(request);
		if (request->RG_ID==-1||len<=0)
		{
			printf("recv the incorrect RG_ID[%ld],len:%d\n",request->RG_ID,len);
			break;
		}

		printf("recv the request of archiving for RG_ID[%ld]\n",request->RG_ID);
		char** originalBlockData=(char**)malloc(sizeof(char*)*K);
		for ( i= 0; i < K; ++i) originalBlockData[i] = (char*)malloc(sizeof(char)*Block_Size);
		//检索原始数据块数据
		//pthread_rwlock_rdlock(&lock4Pool);
		int ret = retrieveOriginalBlockData(request,originalBlockData);
		if (ret==ERR)
		{
			printf("failed to retrieveOriginalBlockData\n");
			return NULL;
		}
		//pthread_rwlock_unlock(&lock4Pool);
		char** parityBlockData = (char**)malloc(sizeof(char*)*R);
		for ( i= 0; i < R; ++i) parityBlockData[i] = (char*)malloc(sizeof(char)*Block_Size);
		//3.编码，生成校验数据
		encode4ParityBlockData(originalBlockData,parityBlockData);
		//4.发送parityBLock，同样需要使用相应的WorkerAsClient2WorkerConnect中的sock_fd
		Worker2WorkerRequest* worker2WorkerRequest = (Worker2WorkerRequest*)malloc(sizeof(Worker2WorkerRequest));
		for (i = 0; i < R; ++i)
		{
			if (nodeID!=request->parityNodeIDs[i])
			{
				worker2WorkerRequest->requestID=(long)rand();
				worker2WorkerRequest->blockID=request->parityIDs[i];
				worker2WorkerRequest->toNodeID=request->parityNodeIDs[i];
				worker2WorkerRequest->fromNodeID=nodeID;
				worker2WorkerRequest->flag=1;
				//发送request
				int isinsamerack = isInSameRack(worker2WorkerRequest->toNodeID,worker2WorkerRequest->fromNodeID);
				int fd=-1;
				if (isinsamerack==1)
				{
					worker2worker_send(worker2WorkerRequest);
					fd=lookforworker2workerfdByNodeID(worker2WorkerRequest->toNodeID);
				}else{
					worker2gateway_send(worker2WorkerRequest);
					fd=worker2gateway_sockfd;
				}
				send_bytes(fd,parityBlockData[i],Block_Size);
				printf("the request for setting parity data blockID[%lld] has been sent\n",worker2WorkerRequest->blockID);
				long requestID4Check;
				recv_bytes(fd,(char*)&requestID4Check,sizeof(long));
				if (worker2WorkerRequest->requestID!=requestID4Check) return NULL;
				printf("the request for setting parity data blockID[%lld] has been done\n",worker2WorkerRequest->blockID);
			}else{
				Block* block = (Block*)malloc(sizeof(Block));
				block->blockID=request->parityIDs[i];
				memcpy(block->dataBuf,parityBlockData[i],sizeof(char)*Block_Size);
				char* key = lltoa(block->blockID);
				cuckoo_insert(pool,key,(void*)block);
				free(key);
			}
			
		}
		//5.删除originalBlock，同样需要使用相应的WorkerAsClient2WorkerConnect中的sock_fd
		for (i = 0; i < K; ++i)
		{
			if (request->originalNodeIDs[i]==nodeID)
			{
				char* key = lltoa(request->blockIDs[i]);
				Block* temp = (Block*)cuckoo_remove(pool,key);
				if (temp!=NULL) free(temp);
				free(key);
				printf("local delete the og data blockID[%lld]\n",worker2WorkerRequest->blockID);
			}else{
				worker2WorkerRequest->requestID=(long)rand();
				worker2WorkerRequest->blockID=request->blockIDs[i];
				worker2WorkerRequest->toNodeID=request->originalNodeIDs[i];
				worker2WorkerRequest->fromNodeID=nodeID;
				worker2WorkerRequest->flag=2;

				int isinsamerack = isInSameRack(worker2WorkerRequest->toNodeID,worker2WorkerRequest->fromNodeID);
				int fd=-1;
				if (isinsamerack==1)
				{
					worker2worker_send(worker2WorkerRequest);
					fd=lookforworker2workerfdByNodeID(worker2WorkerRequest->toNodeID);
				}else{
					worker2gateway_send(worker2WorkerRequest);
					fd=worker2gateway_sockfd;
				}
				printf("the request for deleting og data blockID[%lld] has been sent\n",worker2WorkerRequest->blockID);

				long requestID4Check;
				recv_bytes(fd,(char*)&requestID4Check,sizeof(long));
				if (worker2WorkerRequest->requestID!=requestID4Check) return NULL;
				
				printf("the request for deleting parity data blockID[%lld] has been done\n",worker2WorkerRequest->blockID);
			}
		}
		//6.通过master2worker_sockfd发送response
		send_bytes(master2worker_sockfd,(char*)&request->RG_ID,sizeof(long));
		printf("the response for RG_ID[%ld] has been sent\n",request->RG_ID);
		printf("finish the work of archiving for RG_ID[%ld]\n",request->RG_ID);
		
		for ( i= 0; i < K; ++i) free(originalBlockData[i]);
		free(originalBlockData);
		for (i = 0; i < R; ++i) free(parityBlockData[i]);
		free(parityBlockData);
		if (request) free(request);
		if (worker2WorkerRequest) free(worker2WorkerRequest);
	}
	return NULL;
}

//检索编码所需要的K个数据块数据
int retrieveOriginalBlockData(ArchivalRequest* archivalRequest,char** originalBlockData){
	int i,j;
	Worker2WorkerRequest* worker2WorkerRequest = (Worker2WorkerRequest*)malloc(sizeof(Worker2WorkerRequest));
	for (i = 0; i < K; ++i)
	{
		//分别给其他节点发送get数据块数据的request
		if (archivalRequest->originalNodeIDs[i]==nodeID)
		{
			//不需要检索，数据块在本地
			char* key = lltoa(archivalRequest->blockIDs[i]);
			memcpy(originalBlockData[i],((Block*)cuckoo_get(pool,key))->dataBuf,Block_Size);
			free(key);			
		}else{
			worker2WorkerRequest->requestID=(long)rand();
			worker2WorkerRequest->toNodeID=archivalRequest->originalNodeIDs[i];
			worker2WorkerRequest->fromNodeID=nodeID;
			worker2WorkerRequest->blockID=archivalRequest->blockIDs[i];
			worker2WorkerRequest->flag=0;//get action

			int isinsamerack = isInSameRack(worker2WorkerRequest->toNodeID,worker2WorkerRequest->fromNodeID);
			int fd=-1;
			if (isinsamerack==1)
			{
				worker2worker_send(worker2WorkerRequest);
				fd = lookforworker2workerfdByNodeID(worker2WorkerRequest->toNodeID);
				
			}else{
				worker2gateway_send(worker2WorkerRequest);
				fd=worker2gateway_sockfd;
			}
			printf("the request for retrieving blockdata blockID[%lld] has been sent\n",worker2WorkerRequest->blockID);
			recv_bytes(fd,originalBlockData[i],Block_Size);
			printf("the data for retrieving blockdata blockID[%lld] has been recvived\n",archivalRequest->blockIDs[i]);
			long requestID4Check;
			recv_bytes(fd,(char*)&requestID4Check,sizeof(long));
			
			if (worker2WorkerRequest->requestID!=requestID4Check) return ERR;
		}
	}

	if (worker2WorkerRequest!=NULL) free(worker2WorkerRequest);
	printf("retrieveOriginalBlockData has been done successfully\n");
	return 0;
}


int isInSameRack(int nodeID1,int nodeID2){
	if (nodes[nodeID1].rackID==nodes[nodeID2].rackID)
	{
		return 1;
	}else{
		return 0;
	}
}

int lookforworker2workerfdByNodeID(int targetNodeID){
	int fd;
	for(int j=0;j<NodeNumTotal-2;j++){
		if(workerAsClient2WorkerConnect[j].targetNode==targetNodeID){
			fd=workerAsClient2WorkerConnect[j].sockfd;
			break;
		}
	}
	return fd;
}


void worker2worker_send(Worker2WorkerRequest *request){
	size_t needSend =sizeof(Worker2WorkerRequest);
	char *sendBuf=(char *)malloc(needSend);
	memcpy(sendBuf,request,needSend);
	int fd=lookforworker2workerfdByNodeID(request->toNodeID);   
	send_bytes(fd,sendBuf,needSend);
}


void worker2gateway_send(Worker2WorkerRequest *request){
	size_t needSend =sizeof(Worker2WorkerRequest);
	char *sendBuf=(char *)malloc(needSend);
	memcpy(sendBuf,request,needSend); 
	send_bytes(worker2gateway_sockfd,sendBuf,needSend);
}




void encode4ParityBlockData(char** originalBlockData,char** parityBlockData){
	int* matrix= cauchy_good_general_coding_matrix(K,R,W);
	int* bitmatrix = jerasure_matrix_to_bitmatrix(K,R,W,matrix);
	int** schedule = jerasure_smart_bitmatrix_to_schedule(K,R,W,bitmatrix);
	jerasure_schedule_encode(K,R,W,schedule,originalBlockData,parityBlockData,Block_Size,8);
}

void* worker2Worker_thread(void* arg){
	int sourceNode = (int)arg;
	int isfinish=0;
	while(isfinish==0){
		int ret = handlerRequestIOFromWorker(sourceNode);
		if (ret==ERR)
		{
			isfinish=1;
		}
	}
	printf("worker2Worker_thread has end\n");
	return NULL;
}

void* process4client_thread(void* arg){
	int sockfd = (int)arg;
	int isfinish=0;
	while(isfinish==0){
		int ret = handlerRequestIOFromClient(sockfd);
		if (ret==ERR)
		{
			isfinish=1;
			//isDestory=1;
		}
	}
	printf("process4client_thread has end\n");
	return NULL;
}

void* process4gateway_thread(void* arg){
	int isfinish=0;
	while(isfinish==0){
		int ret = handlerRequestIOFromGateWay();
		if (ret==ERR)
		{
			isfinish=1;
			isDestory=1;
		}
	}
	printf("process4gateway_thread has end\n");
	return NULL;
}

int main()
{
	int ret;
	ret = init_worker();
	if (ret==ERR){
		printf("fail to init worker\n");
		destory_worker();
		return 0;
	}
	pthread_t thread4clients[ClientNum];
	//为client创建服务
	for (int i = 0; i < ClientNum; ++i)
	{
		pthread_create(&thread4clients[i],NULL,process4client_thread,(void*)i);
	}
	//为master创建线程 处理归档
	pthread_t archival_thread_ID;
	pthread_create(&archival_thread_ID,NULL,archival_thread,NULL);

	//为worker创建线程
	pthread_t thread4workers[NodeNumTotal-1];
	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		if (nodeID!=i)
		{
			pthread_create(&thread4workers[i],NULL,worker2Worker_thread,(void*)i);
		}
	}

	pthread_t thread4gateway;
	pthread_create(&thread4gateway,NULL,process4gateway_thread,NULL);


	while(1){
		sleep(1);
		if(isDestory==1){
			break;
		}
	}
	
	for (int i = 0; i < ClientNum; ++i)
	{
		pthread_cancel(thread4clients[i]);
	}

	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		if (nodeID!=i)
		{
			pthread_cancel(thread4workers[i]);
		}
	}

	pthread_cancel(archival_thread_ID);
	pthread_cancel(thread4gateway);
	destory_worker();
	return 0;
}

